package com.alarm.ecp;

import com.alarm.ecp.pojo.AlertEvent;
import com.alarm.ecp.pojo.PayEvent;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

/**
 * 找出那些下单后 10 分钟内没有支付的订单
 */
public class PayAlarmFlink {
    public static void main(String[] args) throws Exception
    {
        // 设置匹配
        Pattern<PayEvent, PayEvent> pattern = Pattern.<PayEvent>
                        begin("begin")
                .where(new IterativeCondition<PayEvent>() {
                    @Override
                    public boolean filter(PayEvent payEvent, Context context) throws Exception {
                        return payEvent.getAction().equals("create");
                    }
                })
                .next("next")
                .where(new IterativeCondition<PayEvent>() {
                    @Override
                    public boolean filter(PayEvent payEvent, Context context) throws Exception {
                        return payEvent.getAction().equals("pay");
                    }
                })
                .within(Time.seconds(600));

        // 定义侧输出流标签
        OutputTag<PayEvent> orderTimeoutOutput = new OutputTag<PayEvent>("orderTimeout") {};

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStream<PayEvent> source = env.fromElements(
                new PayEvent(1L, "create", 1597905234000L),
                new PayEvent(1L, "pay", 1597905235000L),
                new PayEvent(2L, "create", 1597905236000L),
                new PayEvent(2L, "pay", 1597905237000L),
                new PayEvent(3L, "create", 1597905239000L)
        ).assignTimestampsAndWatermarks(
                WatermarkStrategy.<PayEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner((event, timestamp) -> event.getTimeStamp())
        ).keyBy((KeySelector<PayEvent, Object>) PayEvent::getUserId);

        PatternStream<PayEvent> patternStream = CEP.pattern(source, pattern);

        SingleOutputStreamOperator<PayEvent> result = patternStream.select(orderTimeoutOutput, (PatternTimeoutFunction<PayEvent, PayEvent>) (map, l) -> map.get("begin").get(0),
                (PatternSelectFunction<PayEvent, PayEvent>) map -> map.get("begin").get(0));
        DataStream<PayEvent> sideOutput = result.getSideOutput(orderTimeoutOutput);

        result.print("result");
        sideOutput.print("sideOutput");

        SingleOutputStreamOperator<AlertEvent> alarmStream = sideOutput.map(payEvent -> new AlertEvent(payEvent.getUserId().toString(), "10分钟内没有支付的订单！"));

        alarmStream.printToErr();

//        result.printToErr("main stream");
//        sideOutput.printToErr("orderTimeoutOutput stream");
        env.execute("execute cep");
    }
}
